////
// @file block.h
// @brief
// 定义block
// block是记录、索引的存储单元。在MySQL和HBase中，存储单元与分配单位是分开的，一般来说，
// 最小分配单元要比block大得多。
// block的布局如下，每个slot占用2B，这要求block最大为64KB。由于记录和索引要求按照4B对
// 齐，BLOCK_DATA、BLOCK_TRAILER也要求8B对齐。
//
// +--------------------+ <--- 0
// |   common header    |
// +--------------------+ <--- sizeof(CommonHeader)
// |                    |
// |  data/index header | SuperHeader / DataHeader(MetaHeader) / IdleHeader
// |                    |
// +--------------------+ <--- sizeof(SuperHeader) / sizeof(DataHeader) / sizeof(IdleHeader)
// |                    |
// |     data/index     |
// |                    |
// +--------------------+ <--- CommonHeader.freeSpace / getFreeSpaceSize()
// |    (new record)    | ^
// |                    | |
// |     free space     | | getFreeSize() / DataHeader.freeSize
// |                    | |
// |     (new slot)     | v
// +--------------------+ <--- getSlotsPointer() = BLOCK_SIZE - sizeof(unsigned int)[checksum] - block.getSlots() * sizeof(Slot)
// |                    | ^ ^
// |                    | | |
// |       slots        | | getSlots() * sizeof(Slot) / DataHeader.slots * sizeof(Slot)
// |                    | | | getTrailerSize()
// |                    | v |
// +--------------------+ <--- BLOCK_SIZE - sizeof(Trailer) [包括1个Slot]
// | trailer(checksum)  |   v
// +--------------------+ <--- BLOCK_SIZE
//
// @author niexw
// @email niexiaowen@uestc.edu.cn
//
#ifndef __DB_BLOCK_H__
#define __DB_BLOCK_H__

#include "./checksum.h"
#include "./endian.h"
#include "./timestamp.h"
#include "./record.h"
#include "./schema.h"
#include "./datatype.h"

namespace db {

constexpr unsigned short BLOCK_TYPE_IDLE = 0;  // 空闲
constexpr unsigned short BLOCK_TYPE_SUPER = 1; // 超块
constexpr unsigned short BLOCK_TYPE_DATA = 2;  // 数据
constexpr unsigned short BLOCK_TYPE_META = 3;  // 元数据
constexpr unsigned short BLOCK_TYPE_LOG = 4;   // wal日志
constexpr unsigned short BLOCK_TYPE_BTREE_INNER = 5; // btree内部节点
constexpr unsigned short BLOCK_TYPE_BTREE_LEAF = 6;  // btree叶子节点

constexpr unsigned int SUPER_SIZE = 1024 * 4;  // 超块大小为4KB
constexpr unsigned int BLOCK_SIZE = 1024 * 16; // 一般块大小为16KB

#if BYTE_ORDER == LITTLE_ENDIAN
static constexpr int MAGIC_NUMBER = 0x31306264; // magic number
#else
static const int MAGIC_NUMBER = 0x64623031; // magic number
#endif

// TODO: LSN
// 公共头部
struct CommonHeader
{
    unsigned int magic;       // magic number(4B)
    unsigned int spaceid;     // 表空间id(4B)
    unsigned short type;      // block类型(2B)
    unsigned short freespace; // 空闲记录链表(2B)
};

// slots结构
struct Slot
{
    unsigned short offset; // 记录偏移量
    unsigned short length; // 记录大小
};

// 尾部
// READING: 为什么要显式分配一个Slot的空间?
struct Trailer
{
    Slot slots[1];         // slots数组
    unsigned int checksum; // 校验和(4B)
};

// 超块头部
struct SuperHeader : CommonHeader
{
    unsigned int first;      // 第1个数据块(4B)
    long long stamp;         // 时戳(8B)
    unsigned int idle;       // 空闲块(4B)
    unsigned int datacounts; // 数据块个数
    unsigned int idlecounts; // 空闲块个数
    unsigned int self;       // 本块id(4B)
    unsigned int maxid;      // 最大的blockid(4B)
    unsigned int pad;        // 填充位(4B)
    long long records;       // 记录数目(8B)
    unsigned int firstIndex; // 第1个索引块(4B)
};

// 空闲块头部
struct IdleHeader : CommonHeader
{
    unsigned int next; // 后继指针(4B)
};

// 数据块头部
// READING: 为什么要有free size字段?
struct DataHeader : CommonHeader
{
    unsigned int next;       // 下一个数据块(4B)
    long long stamp;         // 时戳(8B)
    unsigned short slots;    // slots[]长度(2B)
    unsigned short freesize; // 空闲空间大小(2B)
    unsigned int self;       // 本块id(4B)
};

// 元数据块头部
using MetaHeader = DataHeader;

struct BTreeHeader: CommonHeader
{
    // 为了保证B+树的结构，next指针和结尾数据存在一起
    // 保证BTreeHeader与DataHeader有相同的结构
    unsigned int flag;       // 下一个数据块(4B)
    long long stamp;         // 时戳(8B)
    unsigned short size;    // slots[]长度(2B)
    unsigned short freeSize; // 空闲空间大小(2B)
    unsigned int self;       // 本块id(4B)
};

struct BTreeNode
{
    unsigned int data;
    Slot key;
};

struct BTreeTrailer
{
    unsigned int data;
    unsigned int checksum; // 校验和(4B)
};

////
// @brief
// 公共block
//
class Block
{
  public:
    unsigned char *buffer_; // block对应的buffer

  public:
    Block()
        : buffer_(nullptr)
    {}

    // 关联buffer
    inline void attach(unsigned char *buffer) { buffer_ = buffer; }
    inline void detach() { buffer_ = nullptr; }

    // 设定magic
    inline void setMagic() const
    {
        auto header = reinterpret_cast<CommonHeader *>(buffer_);
        header->magic = MAGIC_NUMBER;
    }
    inline int getMagic() const
    {
        auto header = reinterpret_cast<CommonHeader *>(buffer_);
        return header->magic;
    }

    // 获取表空间id
    inline unsigned int getSpaceId() const
    {
        auto header = reinterpret_cast<SuperHeader *>(buffer_);
        return be32toh(header->spaceid);
    }
    // 设定表空间id
    inline void setSpaceId(const unsigned int spaceId) const
    {
        auto header = reinterpret_cast<SuperHeader *>(buffer_);
        header->spaceid = htobe32(spaceId);
    }

    // 获取类型
    inline unsigned short getType() const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        return be16toh(header->type);
    }
    // 设定类型
    inline void setType(const unsigned short type) const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        header->type = htobe16(type);
    }

    // 获取free space
    inline unsigned short getFreeSpace() const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        return be16toh(header->freespace);
    }
};

////
// @brief
// 超块
//
class SuperBlock : public Block
{
  public:
    // 关联buffer 直接继承Block的attach与deattach方法
    // inline void attach(unsigned char *buffer) { buffer_ = buffer; }
    // 清超块
    void clear(const unsigned short spaceId) const;

    // 获取第1个数据块
    inline unsigned int getFirst() const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        return be32toh(header->first);
    }
    // 设定数据块链头
    inline void setFirst(const unsigned int first) const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        header->first = htobe32(first);
    }

    // 获取空闲块
    inline unsigned int getIdle() const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        return be32toh(header->idle);
    }
    // 设定空闲块链头
    inline void setIdle(const unsigned int idle) const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        header->idle = htobe32(idle);
    }

    // 获取最大blockid
    inline unsigned int getMaxId() const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        return be32toh(header->maxid);
    }
    // 设定最大blockid
    inline void setMaxId(const unsigned int maxId) const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        header->maxid = htobe32(maxId);
    }

    // 获取时戳
    inline TimeStamp getTimeStamp() const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        TimeStamp ts;
        memcpy(&ts, &header->stamp, sizeof(TimeStamp));
        *reinterpret_cast<long long *>(&ts) = be64toh(*reinterpret_cast<long long *>(&ts));
        return ts;
    }
    // 设定时戳
    inline void setTimeStamp() const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        TimeStamp ts;
        ts.now();
        *reinterpret_cast<long long *>(&ts) = htobe64(*reinterpret_cast<long long *>(&ts));
        memcpy(&header->stamp, &ts, sizeof(TimeStamp));
    }

    // 设置data counts
    inline void setDataCounts(unsigned int counts) const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        header->datacounts = htobe32(counts);
    }
    // 获取data counts
    inline unsigned int getDataCounts() const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        return be32toh(header->datacounts);
    }

    // 设置idle counts
    inline void setIdleCounts(unsigned int counts) const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        header->idlecounts = htobe32(counts);
    }
    // 获取data counts
    inline unsigned int getIdleCounts() const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        return be32toh(header->idlecounts);
    }

    // 设置self
    inline void setSelf() const
    {
        auto  header = reinterpret_cast<MetaHeader *>(buffer_);
        header->self = htobe32(0);
    }
    // 获取self
    inline unsigned int getSelf() const
    {
        auto  header = reinterpret_cast<MetaHeader *>(buffer_);
        return be32toh(header->self);
    }

    // 设定checksum
    inline void setChecksum() const
    {
        auto  trailer =
            reinterpret_cast<Trailer *>(buffer_ + SUPER_SIZE - sizeof(Trailer));
        trailer->checksum = 0; // 先要清0，以防checksum计算在内
        trailer->checksum = checksum32(buffer_, SUPER_SIZE);
    }
    // 获取checksum
    inline unsigned int getChecksum() const
    {
        auto  trailer =
            reinterpret_cast<Trailer *>(buffer_ + SUPER_SIZE - sizeof(Trailer));
        return trailer->checksum;
    }
    // 检验checksum
    inline bool checksum() const
    {
        unsigned int sum = 0;
        sum = checksum32(buffer_, SUPER_SIZE);
        return !sum;
    }
    // 设定空闲链头
    inline void setFreeSpace(const unsigned short freeSpace) const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        header->freespace = htobe16(freeSpace);
    }

    inline void setRecords(const long long s) const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        header->records = htobe64(s);
    }
    inline long long getRecords() const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        return be64toh(header->records);
    }
    inline void setFirstIndex(const unsigned int firstIndex) const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        header->firstIndex = htobe32(firstIndex);
    }
    inline unsigned int getFirstIndex() const
    {
        auto  header = reinterpret_cast<SuperHeader *>(buffer_);
        return be32toh(header->firstIndex);
    }
};

////
// @brief
// 数据块
//
class MetaBlock : public Block
{
  public:
    // 清数据块
    void clear(const unsigned short spaceId, const unsigned int self, const unsigned short type) const;

    // 获取空闲块
    inline unsigned int getNext() const
    {
        auto  header = reinterpret_cast<MetaHeader *>(buffer_);
        return be32toh(header->next);
    }
    // 设定block链头
    inline void setNext(const unsigned int next) const
    {
        auto  header = reinterpret_cast<MetaHeader *>(buffer_);
        header->next = htobe32(next);
    }

    // 获取时戳
    inline TimeStamp getTimeStamp() const
    {
        auto  header = reinterpret_cast<MetaHeader *>(buffer_);
        TimeStamp ts;
        ts.retrieve(header->stamp);
        return ts;
    }
    // 设定时戳
    inline void setTimeStamp() const
    {
        auto  header = reinterpret_cast<MetaHeader *>(buffer_);
        TimeStamp ts;
        ts.now();
        ts.store(&header->stamp);
    }

    // 获取空闲空间大小
    inline unsigned short getFreeSize() const
    {
        auto  header = reinterpret_cast<MetaHeader *>(buffer_);
        return be16toh(header->freesize);
    }
    // 设定空闲空间大小
    inline void setFreeSize(const unsigned short size) const
    {
        auto  header = reinterpret_cast<MetaHeader *>(buffer_);
        header->freesize = htobe16(size);
    }

    // 设置slots数目
    inline void setSlots(const unsigned short slots) const
    {
        auto  header = reinterpret_cast<MetaHeader *>(buffer_);
        header->slots = htobe16(slots);
    }
    // 获取slots数目
    inline unsigned short getSlots() const
    {
        auto  header = reinterpret_cast<MetaHeader *>(buffer_);
        return be16toh(header->slots);
    }

    // 设置self
    inline void setSelf(const unsigned int id) const
    {
        auto  header = reinterpret_cast<MetaHeader *>(buffer_);
        header->self = htobe32(id);
    }
    // 获取self
    inline unsigned int getSelf() const
    {
        auto  header = reinterpret_cast<MetaHeader *>(buffer_);
        return be32toh(header->self);
    }

    // 设定checksum
    inline void setChecksum() const
    {
        auto  trailer =
            reinterpret_cast<Trailer *>(buffer_ + BLOCK_SIZE - sizeof(Trailer));
        trailer->checksum = 0; // 先要清0，防止checksum计算出错
        trailer->checksum = checksum32(buffer_, BLOCK_SIZE);
    }
    // 获取checksum
    inline unsigned int getChecksum() const
    {
        auto  trailer =
            reinterpret_cast<Trailer *>(buffer_ + BLOCK_SIZE - sizeof(Trailer));
        return trailer->checksum;
    }
    // 检验checksum
    inline bool checksum() const
    {
        unsigned int sum = 0;
        sum = checksum32(buffer_, BLOCK_SIZE);
        return !sum;
    }

    // 获取trailer大小
    inline unsigned short getTrailerSize() const
    {
        auto  header = reinterpret_cast<MetaHeader *>(buffer_);
        return ALIGN_TO_SIZE(
            be16toh(header->slots) * sizeof(Slot) + sizeof(unsigned int));
        // READING: 计算trailer的大小，slots[]的大小+checksum的大小
    }
    // 获取slots[]指针
    inline Slot *getSlotsPointer() const
    {
        auto  header = reinterpret_cast<MetaHeader *>(buffer_);
        return reinterpret_cast<Slot *>(
            buffer_ + BLOCK_SIZE - sizeof(unsigned int) -
            be16toh(header->slots) * sizeof(Slot));
    }
    // 获取frees pace空间大小
    inline unsigned short getFreeSpaceSize() const
    {
        auto  header = reinterpret_cast<MetaHeader *>(buffer_);
        return BLOCK_SIZE - getTrailerSize() - be16toh(header->freespace);
    }
    // 设定free space偏移量
    inline void setFreeSpace(unsigned short freeSpace) const
    {
        auto  header = reinterpret_cast<MetaHeader *>(buffer_);
        // 判断是不是超过了Trailer的界限
        unsigned short upper = BLOCK_SIZE - getTrailerSize();
        if (freeSpace >= upper) freeSpace = 0; //超过界限则设置为0
        header->freespace = htobe16(freeSpace);
    }

    // 分配一个空间，直接返回指针，后续需要重新排列slots[]，second表示是否需要reorder
    std::pair<unsigned char *, bool>
    allocate(unsigned short space, unsigned short index) const;
    // 给定一条记录的槽位下标，回收一条记录，回收slots[]中分配的槽位
    void deallocate(unsigned short index) const;
    // 回收删除记录的资源，回收了slots资源，后续需要重排slots[]
    void shrink() const;
    // 对slots[]重排
    inline void reorder(const DataType *type, const unsigned int key) const
    {
        type->sort(buffer_, key);
    }

    // 引用slots[]
    bool refSlots(const unsigned short index, Record &record) const
    {
        if (buffer_ == nullptr || index >= getSlots()) return false;

        Slot *slots = getSlotsPointer();
        record.attach(
            buffer_ + be16toh(slots[index].offset),
            be16toh(slots[index].length));
        return true;
    }
};

////
// @brief
// DataBlock直接从MetaBlock派生
//
class Table;
class DataBlock : public MetaBlock
{
  public:
    struct RecordIterator
    {
        DataBlock *block;
        Record record;
        unsigned short index;

        RecordIterator();
        ~RecordIterator();
        RecordIterator(const RecordIterator &other);

        // 前置操作
        RecordIterator &operator++();
        // 后置操作
        RecordIterator operator++(int);
        // 前置操作
        RecordIterator &operator--();
        // 后置操作
        RecordIterator operator--(int);
        // 加减操作
        RecordIterator &operator+=(int);
        RecordIterator &operator-=(int);
        // 数据块指针
        Record *operator->();
    };

  public:
    Table *table_; // 指向table

  public:
    DataBlock()
        : table_(nullptr)
    {}
    explicit DataBlock(Table *table):
        table_(table)
    {}

    // 设定table
    inline void setTable(Table *table) { table_ = table; }
    // 获取table
    inline Table *getTable() const { return table_; }

    // 查询记录
    // 给定一个关键字，从slots[]上搜索到该记录：
    // 1. 根据meta确定key的位置；
    // 2. 采用二分查找在slots[]上寻找
    // 返回值：
    // 返回lowerbound
    unsigned short searchRecord(void *buf, size_t len) const;
    // 插入记录
    // 在block中插入记录，步骤如下：
    // 1. 先检查空间是否足够，如果够，则插入，然后重新排序；
    // 2. 不够，根据key寻找插入位置，从这个位置将block劈开；
    // 3. 计算劈开后前面序列的长度，然后判断新记录加入后空间是否足够，够则插入；
    // 4. 先将新的记录插入一个新的block，然后挪动原有记录到新的block；
    // 返回值：
    // true - 表示记录完全插入
    // false - 表示block被分裂
    std::pair<bool, unsigned short>
    insertRecord(const std::vector<struct iovec> &iov) const;
    // TODO: 修改记录
    // 修改一条存在的记录
    // 先标定原记录为tombstone，然后插入新记录
    bool updateRecord(const std::vector<struct iovec> &iov) const;
    // TODO: 分裂块位置
    // 给定新增的记录大小和位置，计算从何处开始分裂该block
    // 1. 先按照键排序
    // 2. 从0开始枚举所有记录，累加长度，何时超过一半，即为分裂位置
    std::pair<unsigned short, bool>
    splitPosition(size_t space, unsigned short index) const;
    // 拷贝一条记录
    // 如果新block空间不够，简单地返回false
    bool copyRecord(const Record &record) const;
    /*
     * 引用一条记录,是searchRecord的封装
     */
    std::pair<bool, unsigned short> refRecord(Record &record, void *keyBuf, unsigned int len) const;
    /*
     * 删除一条记录,是dealloc的封装
     */
    bool removeRecord(void *keyBuf, unsigned int len) const;

    // 记录分配长度
    unsigned short requireLength(const std::vector<struct iovec> &iov) const;

    RecordIterator beginRecord();
    RecordIterator endRecord();
};


class BTreeBlock: public Block
{
  public:
    Table *table_=nullptr;
  public:
    BTreeBlock()
        : table_(nullptr) {};
    BTreeBlock(Table *table):
        table_(table)
    {}
    inline void setTable(Table *table = nullptr) { table_ = table; }
    void clear(const unsigned short spaceId, const unsigned int self);

    inline unsigned int getFlag() const
    {
        auto  header = reinterpret_cast<BTreeHeader *>(buffer_);
        return be32toh(header->flag);
    }
    inline void setFlag(const unsigned int flag) const
    {
        auto  header = reinterpret_cast<BTreeHeader *>(buffer_);
        header->flag = htobe32(flag);
    }
    inline TimeStamp getTimeStamp() const
    {
        auto  header = reinterpret_cast<BTreeHeader *>(buffer_);
        TimeStamp ts;
        ts.retrieve(header->stamp);
        return ts;
    }
    inline void setTimeStamp() const
    {
        auto  header = reinterpret_cast<BTreeHeader *>(buffer_);
        TimeStamp ts;
        ts.now();
        ts.store(&header->stamp);
    }

    inline unsigned short getFreeSize() const
    {
        auto  header = reinterpret_cast<BTreeHeader *>(buffer_);
        return be16toh(header->freeSize);
    }
    inline void setFreeSize(const unsigned short size) const
    {
        auto  header = reinterpret_cast<BTreeHeader *>(buffer_);
        header->freeSize = htobe16(size);
    }

    inline unsigned short getSize() const
    {
        auto  header = reinterpret_cast<BTreeHeader *>(buffer_);
        return be16toh(header->size);
    }
    inline void setSize(const unsigned short size) const
    {
        auto  header = reinterpret_cast<BTreeHeader *>(buffer_);
        header->size = htobe16(size);
    }
    inline unsigned int getSelf() const
    {
        auto  header = reinterpret_cast<BTreeHeader *>(buffer_);
        return be32toh(header->self);
    }
    inline void setSelf(const unsigned int self) const
    {
        auto  header = reinterpret_cast<BTreeHeader *>(buffer_);
        header->self = htobe32(self);
    }
    // 设定checksum
    inline void setChecksum() const
    {
        auto  trailer =
            reinterpret_cast<BTreeTrailer *>(buffer_ + BLOCK_SIZE - sizeof(BTreeTrailer));
        trailer->checksum = 0; // 先要清0，防止checksum计算出错
        trailer->checksum = checksum32(buffer_, BLOCK_SIZE);
    }
    inline bool checksum() const
    {
        unsigned int sum = 0;
        sum = checksum32(buffer_, BLOCK_SIZE);
        return !sum;
    }

    inline unsigned short getTrailerSize() const
    {
        auto header = reinterpret_cast<BTreeHeader *>(buffer_);
        return ALIGN_TO_SIZE(
            be16toh(header->size) * sizeof(BTreeNode) + sizeof(BTreeTrailer));
    }

    // 获取free space空间大小
    inline unsigned short getFreeSpaceSize() const
    {
        auto header = reinterpret_cast<BTreeHeader *>(buffer_);
        return BLOCK_SIZE - getTrailerSize() - be16toh(header->freespace);
    }
    // 设定free space偏移量
    inline void setFreeSpace(unsigned short freeSpace) const
    {
        auto header = reinterpret_cast<BTreeHeader *>(buffer_);
        // 判断是不是超过了Trailer的界限
        unsigned short upper = BLOCK_SIZE - getTrailerSize();
        if (freeSpace >= upper) freeSpace = 0; //超过界限则设置为0
        header->freespace = htobe16(freeSpace);
    }

    inline BTreeNode *getNodesPointer() const
    {
        if (buffer_ == nullptr) return nullptr;
        return reinterpret_cast<BTreeNode *>(
            buffer_ + BLOCK_SIZE - sizeof(BTreeTrailer) -
            getSize() * sizeof(BTreeNode));
    }
    inline unsigned char *getRecordsPointer() const
    {
        if (buffer_ == nullptr) return nullptr;
        return buffer_ + sizeof(BTreeHeader);
    }
    inline BTreeTrailer *getTrailerPointer() const
    {
        if (buffer_ == nullptr) return nullptr;
        return reinterpret_cast<BTreeTrailer *>(buffer_ + BLOCK_SIZE - sizeof(BTreeTrailer));
    }
    inline Slot getKeySlot(const unsigned short index) const
    {
        if (buffer_ == nullptr || index > getSize()) return {};
        BTreeNode *node = getNodesPointer();
        return {be16toh(node[index].key.offset), be16toh(node[index].key.length)};
    }
    inline bool getKey(const unsigned short index, void * keyBuf, unsigned int *keyLen) const
    {
        if (buffer_ == nullptr || index >= getSize()) return false;
        BTreeNode *node = getNodesPointer();
        *keyLen = be16toh(node[index].key.length);
        if (*keyLen == 0) return false; // key长度为0
        if (keyBuf == nullptr) return true; // keyBuf为nullptr则不拷贝
        memcpy(keyBuf, buffer_ + be16toh(node[index].key.offset), *keyLen);
        return true;
    }
    inline bool refKey(const unsigned short index, void **keyBuf, unsigned int *keyLen) const
    {
        if (buffer_ == nullptr || index >= getSize()) return false;
        BTreeNode *node = getNodesPointer();
        *keyLen = be16toh(node[index].key.length);
        if (*keyLen == 0) return false; // key长度为0
        if (keyBuf == nullptr) return true; // keyBuf为nullptr则不拷贝
        *keyBuf = buffer_ + be16toh(node[index].key.offset);
        return true;
    }
    inline unsigned int getChild(const unsigned short index) const
    {
        if (buffer_ == nullptr || index > getSize()) return 0;
        BTreeNode *node = getNodesPointer();
        return be32toh(node[index].data);
    }
    inline bool updateKey(const unsigned short index, const void *keyBuf, unsigned int keyLen)
    {
        if (buffer_ == nullptr || index >= getSize() || keyBuf == nullptr || keyLen == 0) return false;
        BTreeNode *node = getNodesPointer();
        unsigned short offset = be16toh(node[index].key.offset);
        unsigned short length = be16toh(node[index].key.length);
        if (length >= keyLen) {
            memcpy(buffer_ + offset, keyBuf, keyLen);
            node[index].key.length = htobe16(keyLen);
            return true;
        }
        if (getFreeSize() < keyLen)
            return false;
        memcpy(buffer_ + getFreeSpace(), keyBuf, keyLen);
        // 更新key
        node[index].key.offset = htobe16(getFreeSpace());
        node[index].key.length = htobe16(keyLen);
        // 更新free space
        setFreeSpace(getFreeSpace() + keyLen);
        setFreeSize(getFreeSize() - keyLen);
        return true;
    }

    unsigned char * allocate(unsigned short space, unsigned short index, unsigned int data) const;
    void deallocate(unsigned short index) const;
    void shrink() const;
    inline unsigned short splitPosition() const
    {
        static constexpr unsigned short BlockHalf = (BLOCK_SIZE - sizeof(BTreeHeader) - sizeof(BTreeTrailer)) / 2;
        unsigned short count = getSize();
        size_t half = 0;
        BTreeNode *nodes = getNodesPointer();
        if (nodes == nullptr) {
            return -1;
        }
        unsigned short i;
        for (i = 0; i < count; ++i) {
            half += nodes[i].key.length + sizeof(nodes[0]);
            if (half > BlockHalf) {
                break;
            }
        }
        return i;
    }

    inline bool isLeafNode() const
    {
        return getType() == BLOCK_TYPE_DATA;
    }
    inline unsigned int searchNode(const unsigned int child) const
    {
        if (buffer_ == nullptr) return -1;
        BTreeNode *nodes = getNodesPointer();
        if (nodes == nullptr) return -1;
        unsigned int i = 0;
        for (; i <= getSize(); ++i) {
            if (be32toh(nodes[i].data) == child) {
                return i; // 找到
            }
        }
        return -1; // 没有找到
    }
    unsigned int searchNode(void *keyBuf, size_t keyLen) const;
    inline bool appendKey(const void *keyBuf, const size_t keyLen, const unsigned int child)
    {
        if (buffer_ == nullptr || keyBuf == nullptr || keyLen == 0) return false;
        unsigned char *ptr = allocate(static_cast<unsigned short>(keyLen), 0, child);
        if (ptr == nullptr) return false; // 分配失败
        memcpy(ptr, keyBuf, keyLen);
        return true;
    }

};

inline bool operator==(
    const DataBlock::RecordIterator &x,
    const DataBlock::RecordIterator &y)
{
    if (x.block == nullptr && y.block == nullptr) return true;
    if (x.block == nullptr || y.block == nullptr) return false;
    if (x.block == y.block && x.index == y.index) return true;
    return false;
}
inline bool operator!=(
    const DataBlock::RecordIterator &x,
    const DataBlock::RecordIterator &y)
{
    return !operator==(x, y);
}
// 以下操作要求block相同
inline bool operator<(
    const DataBlock::RecordIterator &x,
    const DataBlock::RecordIterator &y)
{
    // ASSERT(x.block == y.block);
    return x.index < y.index;
}
inline bool operator>(
    const DataBlock::RecordIterator &x,
    const DataBlock::RecordIterator &y)
{
    // ASSERT(x.block == y.block);
    return x.index > y.index;
}
inline bool operator<=(
    const DataBlock::RecordIterator &x,
    const DataBlock::RecordIterator &y)
{
    // ASSERT(x.block == y.block);
    return x.index <= y.index;
}
inline bool operator>=(
    const DataBlock::RecordIterator &x,
    const DataBlock::RecordIterator &y)
{
    // ASSERT(x.block == y.block);
    return x.index >= y.index;
}

} // namespace db

#endif // __DB_BLOCK_H__
